package com.demo.base.rabbitmq.consumer;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.concurrent.Executor;

import static com.demo.base.rabbitmq.constant.MqConstant.DELAY_QUEUE;
import static com.demo.base.rabbitmq.constant.MqConstant.REAL_TIME_QUEUE;

/**
 * @author 曳戈泰尔
 * @version 1.0
 * @description 消息消费端
 * @date 2023/3/10 4:56 PM
 */
@Component
@Slf4j
public class MqConsumer {

    /** 注入 线程池 */
    private Executor threadPoolExecutor;

    @Autowired
    public void setThreadPoolExecutor(Executor threadPoolExecutor) {
        this.threadPoolExecutor = threadPoolExecutor;
    }

    /**
     * 监听消费延时消息
     *
     * @param content
     * @param message
     * @param channel
     */
    @RabbitListener(queues = {DELAY_QUEUE})
    @RabbitHandler
    public void processDelay(String content, Message message, Channel channel) {
        try {
            log.info("延迟队列内容：[{}]", content);

            // 消息的可定确认，第二个参数如果为 true 将一次性确认所有小于 deliveryTag的消息
            // 消息确认
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

            // TODO 处理消息

            log.info("超时消息处理完毕");

        } catch (Exception e) {
            log.error("延迟消息处理失败，内容：{}，原因： {}", content, e.getMessage());

            // 直接拒绝消费该消息
            // 后面的参数要是false，不会重新进入业务队列，会进入死信队列
            // 后面的参数要是true，会重新进入业务队列，不会进入死信队列
            try {
                // 消费失败，重新进入业务队列
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
            } catch (IOException ex) {
                log.error("延迟消息拒绝消费失败，内容：{}，原因：{}", content, e.getMessage());
            }
        }
    }

    /**
     * 消费普通消息 单条消费
     *
     * <p>通过开启线程池和线程睡眠，保证处理频率
     *
     * <p>如果不需要限定消费频率，取消线程池和线程睡眠
     *
     * <p>如果想要高效快速处理，开启线程池
     *
     * @param content
     * @param message
     * @param channel
     */
    @RabbitListener(queues = {REAL_TIME_QUEUE})
    @RabbitHandler
    public void processRealTime(String content, Message message, Channel channel) {
        try {
            log.info("普通队列的内容：[{}]", content);

            // 消息的可定确认，第二个参数如果为true将一次性确认所有小于deliveryTag的消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

            // 开线程处理
            threadPoolExecutor.execute(
                    new Runnable() {
                        @Override
                        public void run() {

                            // TODO 处理消息
                        }
                    });

            // 保持处理频率，设置超时时间，每一次时间都一样
            // 睡眠200ms之后取下一次队列数据
            Thread.sleep(200);

            log.info("普通消息处理完毕，消息内容：{}", content);

        } catch (Exception e) {
            log.error("普通消息处理失败，内容：{}，原因： {}", content, e.getMessage());

            // 直接拒绝消费该消息，
            // 后面的参数要是false，不会重新进入业务队列，会进入死信队列
            // 后面的参数要是true，会重新进入业务队列，不会进入死信队列
            try {
                // 消费失败，重新进入业务队列
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
            } catch (IOException ex) {
                log.error("普通消息拒绝消费失败，内容：{}，原因：{}", content, e.getMessage());
            }
        }
    }
}
